package utils

/**
@auth Crastlin@163.com
@date 2020-10
*/
import (
	"context"
	"fmt"
	"github.com/go-redis/redis/v7"
	"net"
	"strconv"
	"sync"
	"time"
)

type Redis struct {
	*redis.Client
}

var (
	redisInstanceSet map[string]*Redis
	redisOnceSet     map[string]*sync.Once
	redisConfig      RedisConfig
	defaultDb        int
)

func getDatabaseIndex(selectDb ...int) int {
	if redisConfig.Host == "" {
		redisConfig = GetConfig().Redis
		defaultDb = redisConfig.DB
	}

	db := 16
	if len(selectDb) > 0 && selectDb[0] >= 0 {
		db = selectDb[0]
	} else {
		db = defaultDb
	}
	if db < 0 || db > 15 {
		db = redisConfig.DB
	}
	return db
}

// RedisClient get singleton redis
func RedisClient(selectDb ...int) *Redis {
	db := getDatabaseIndex(selectDb...)
	if redisOnceSet == nil {
		redisOnceSet = make(map[string]*sync.Once, 0)
		redisInstanceSet = make(map[string]*Redis, 0)
	}
	key := "db_" + strconv.Itoa(db)
	if client, ok := redisInstanceSet[key]; ok {
		_, err := client.Ping().Result()
		if err == nil {
			return client
		}
	}
	redisOnceSet[key] = &sync.Once{}
	redisOnceSet[key].Do(func() {
		redisClient := &Redis{}
		redisClient.Client = redis.NewClient(&redis.Options{
			Addr:     fmt.Sprintf("%s:%d", redisConfig.Host, redisConfig.Port),
			Password: redisConfig.Passwd,
			DB:       db,
			// 连接池配置
			PoolSize:     redisConfig.PoolSize,
			MinIdleConns: redisConfig.MinIdleConnection,
			// 超时配置
			DialTimeout:  time.Duration(redisConfig.DialTimeout) * time.Second,
			ReadTimeout:  time.Duration(redisConfig.ReadTimeout) * time.Second,
			WriteTimeout: time.Duration(redisConfig.WriteTimeout) * time.Second,
			PoolTimeout:  time.Duration(redisConfig.PoolTimeout) * time.Second,
			//闲置连接检查包括IdleTimeout，MaxConnAge
			IdleCheckFrequency: 60 * time.Second, //闲置连接检查的周期，默认为1分钟，-1表示不做周期性检查，只在客户端获取连接时对闲置连接进行处理。
			IdleTimeout:        5 * time.Minute,  //闲置超时，默认5分钟，-1表示取消闲置超时检查
			MaxConnAge:         0 * time.Second,  //连接存活时长，从创建开始计时，超过指定时长则关闭连接，默认为0，即不关闭存活时长较长的连接
			//命令执行失败时的重试策略
			MaxRetries:      0,                    // 命令执行失败时，最多重试多少次，默认为0即不重试
			MinRetryBackoff: 8 * time.Millisecond, //每次计算重试间隔时间的下限，默认8毫秒，-1表示取消间隔
			MaxRetryBackoff: 512 * time.Millisecond,
			//可自定义连接函数
			Dialer: func(ctx context.Context, network, addr string) (net.Conn, error) {
				netDialer := &net.Dialer{
					Timeout:   5 * time.Second,
					KeepAlive: 5 * time.Minute,
				}
				return netDialer.Dial("tcp", fmt.Sprintf("%s:%d", redisConfig.Host, redisConfig.Port))
			},

			//钩子函数
			OnConnect: func(conn *redis.Conn) error { //仅当客户端执行命令时需要从连接池获取连接时，如果连接池需要新建连接时则会调用此钩子函数
				fmt.Printf("conn=%v\n", conn)
				return nil
			},
		})
		_, err := redisClient.Ping().Result()
		if err != nil {
			fmt.Println(fmt.Sprintf("connect redis server was failed:%v", err.Error()))
		}
		redisInstanceSet[key] = redisClient
	})
	return redisInstanceSet[key]
}

// RedisKey make key with prefix
func RedisKey(key string, prefix string) string {
	return prefix + "_" + key
}

func Publish(channel string, message string) error {
	client := RedisClient()
	result := client.Publish(channel, message)
	if result.Err() != nil {
		fmt.Println("failed to publish message: ", result.Err())
		return result.Err()
	}
	fmt.Println("Publish result: ", result.Val())
	return nil
}

func Subscribe(channel string, runnable func(message string)) {
	client := RedisClient()
	pubSub := client.Subscribe(channel)
	defer pubSub.Close()
	// receive message
	limit := GetConfig().Consumer.Limit
	var wg sync.WaitGroup
	index := 0
	for {
		msg, err := pubSub.ReceiveMessage()
		if err != nil {
			fmt.Println("Failed to receive message: ", err)
			break
		}
		if index == 0 {
			wg.Add(limit)
		}
		fmt.Println("current goroutine index：", index)
		fmt.Println("Subscribe Received message: ", msg.Payload)
		go func(message string) {
			defer wg.Done()
			runnable(message)
		}(msg.Payload)
		index++
		if index >= limit {
			wg.Wait()
			index = 0
		}
	}
}

// RunConsumer 消费队列通用方法
func RunConsumer(mode string, subscribeChannel string, listKey string, limit int, handle func(message string)) {
	if mode == "subscribe" {
		fmt.Println("服务运行中...：")
		Subscribe(subscribeChannel, handle)
	} else {
		redisClient := RedisClient()
		waitTime := time.Duration(10)

		var wg sync.WaitGroup
		index := 0
		for {
			messages, err := redisClient.BRPop(waitTime*time.Second, listKey).Result()
			if err != nil {
				fmt.Println("等待队列....")
				time.Sleep(3 * time.Second)
				continue
			}
			if index == 0 {
				wg.Add(limit)
			}
			for _, message := range messages {
				fmt.Println("BRPOP: ", message)
				if message == listKey {
					continue
				}
				go func(message string) {
					defer wg.Done()
					handle(message)
				}(message)
			}
			index++
			if index >= limit {
				wg.Wait()
				index = 0
			}
		}
	}
}
